package com.raylu.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.raylu.realtime.bean.PageBean;
import com.raylu.realtime.utils.KafkaSinkUtil;
import com.raylu.realtime.utils.KafkaSourceUtil;
import com.raylu.realtime.utils.PropertiesUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Description:
 * <p>访客跳出明细数据</p>
 * <p>跳出率是指仅阅读了一个页面就离开的用户占一组页面或一个页面拜访次数的百分比。跳出次数是指拜访者不拜访您网站的其他任何一页便从进入页退出的次数。</p>
 * Create by lucienoz on 2021/12/29.
 * Copyright © 2021 lucienoz. All rights reserved.
 */
public class UserJumpOutDetailApp {
    public static void main(String[] args) throws Exception {
        Properties load = PropertiesUtil.load("config.properties");
        //TODO 1. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2. 检查点设置
//        env.enableCheckpointing(5000L);
//        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);
//        env.getCheckpointConfig().setCheckpointInterval(3000L);
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(3L)));
//        env.setStateBackend(new FsStateBackend(load.getProperty("user.junmout.app.fsstatebackend.url")));
        //TODO 3. 从Kafka读取主题DWD_PAGE_LOG数据
        KeyedStream<PageBean, String> pageBeanMidKeyedStream = env.addSource(KafkaSourceUtil.getKafkaSource(load.getProperty("base.log.app.kafka.sink-topic1"), load.getProperty("user.junmout.app.kafka.group-id")))
                //TODO 3.1. 将page_log字符串转成PageBean
                .map(r -> JSON.parseObject(r, PageBean.class))
                //TODO 4. 设置读入流的WaterMark
                .assignTimestampsAndWatermarks(WatermarkStrategy.<PageBean>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<PageBean>() {
                            @Override
                            public long extractTimestamp(PageBean element, long recordTimestamp) {

                                return element.getTs();
                            }
                        }))
                //TODO 5. 根据mid进行逻辑分区
                .keyBy(r -> r.getMid());
        //TODO 6. 定义CEP的pattern
        Pattern<PageBean, PageBean> pattern = Pattern.<PageBean>begin("first")
                .where(new SimpleCondition<PageBean>() {
                    @Override
                    public boolean filter(PageBean value) throws Exception {
                        if (value.last_page_id == null || value.last_page_id.length() == 0) {
                            return true;
                        }
                        return false;
                    }
                })
                .next("second")
                .where(new SimpleCondition<PageBean>() {
                    @Override
                    public boolean filter(PageBean value) throws Exception {
                        return true;
                    }
                })
                .within(Time.seconds(10L));
        //TODO 7. 数据通过CEP将按照pattern进行筛选，超时数据为跳出数据通过侧输出进行输出
        SingleOutputStreamOperator<PageBean> resultDS = CEP.pattern(pageBeanMidKeyedStream, pattern).flatSelect(new OutputTag<PageBean>("timeout-pagebean") {
                                                                                                             },
                new PatternFlatTimeoutFunction<PageBean, PageBean>() {
                    @Override
                    public void timeout(Map<String, List<PageBean>> pattern, long timeoutTimestamp, Collector<PageBean> out) throws Exception {
                        List<PageBean> pageBeanList = pattern.get("first");
                        pageBeanList.forEach(out::collect);
                    }
                }, new PatternFlatSelectFunction<PageBean, PageBean>() {
                    @Override
                    public void flatSelect(Map<String, List<PageBean>> pattern, Collector<PageBean> out) throws Exception {
                        //Nothing TODO
                    }
                });

        SingleOutputStreamOperator<String> result = resultDS.getSideOutput(new OutputTag<PageBean>("timeout-pagebean") {
        }).map(JSON::toJSONString);
        result.addSink(KafkaSinkUtil.getKafkaSink(load.getProperty("user.junmout.app.kafka.topic")));
        result.print("跳出>>");
        env.execute();

    }
}
